﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Wfxin.RabbitMQ;

namespace Wfxin.Web.Api.MQ
{
    internal sealed class RabbitMQOrderConsumerClient : IConsumerClient
    {
        private readonly IOrderConnectionChannelPool _connectionChannelPool;
        private readonly string _exchageName;
        private readonly string _exchageType;
        private readonly string _queueName;
        private readonly RabbitMQOptions _rabbitMQOptions;

        private IModel _channel;

        private IConnection _connection;
        private ulong _deliveryTag;
        private IBasicProperties _basicProperties;

        public RabbitMQItem RabbitMQItem
        {
            get
            {
                return _rabbitMQOptions.Items.First(o => o.Name.Equals("CreateOrder"));
            }
        }

        public RabbitMQOrderConsumerClient(string queueName, IOrderConnectionChannelPool connectionChannelPool, RabbitMQOptions options)
        {
            _queueName = queueName;
            _connectionChannelPool = connectionChannelPool;
            _rabbitMQOptions = options;

            _exchageName = connectionChannelPool.Exchange;
            _exchageType = connectionChannelPool.ExchangeType;

            InitClient();
        }

        public event EventHandler<MessageContext> OnMessageReceived;

        public event EventHandler<LogMessageEventArgs> OnLog;

        public string ServersAddress => _rabbitMQOptions.HostName;

        public void Subscribe(IEnumerable<string> topics)
        {
            if (topics == null)
            {
                throw new ArgumentNullException(nameof(topics));
            }

            foreach (var topic in topics)
            {
                _channel.QueueBind(_queueName, _exchageName, topic);
            }
        }

        public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
        {
            var consumer = new EventingBasicConsumer(_channel);
            consumer.Received += OnConsumerReceived;
            consumer.Shutdown += OnConsumerShutdown;
            consumer.Registered += OnConsumerRegistered;
            consumer.Unregistered += OnConsumerUnregistered;
            consumer.ConsumerCancelled += OnConsumerConsumerCancelled;

            _channel.BasicConsume(_queueName, false, consumer);

            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();
                cancellationToken.WaitHandle.WaitOne(timeout);
            }

            // ReSharper disable once FunctionNeverReturns
        }

        public void Commit()
        {
            _channel.BasicAck(_deliveryTag, false);
        }

        public void Reject()
        {
            _channel.BasicReject(_deliveryTag, true);
        }

        public void MoveToReply(byte[] data)
        {
            try
            {
                string exchange = $"{_exchageName}@reply";
                string routingKey = $"{this.RabbitMQItem.RouteKey}@reply";
                if (this.GetRetryCount(_basicProperties) > 5)
                {
                    exchange = $"{_exchageName}@failed";
                    routingKey = $"{this.RabbitMQItem.RouteKey}@failed";
                }
                _channel.BasicPublish(exchange: exchange, routingKey: routingKey, basicProperties: _basicProperties, body: data);
                _channel.BasicReject(_deliveryTag, false);
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        private int GetRetryCount(IBasicProperties props)
        {
            int count = 0;
            if (_basicProperties.Headers != null && _basicProperties.Headers.ContainsKey("x-death"))
            {
                var obj01 = (List<object>)_basicProperties.Headers["x-death"];
                if (obj01.Any())
                {
                    var obj02 = obj01.Cast<Dictionary<string, object>>().FirstOrDefault();
                    if (int.TryParse(obj02["count"].ToString(), out count))
                    {
                        //获取重试次数成功
                    }
                }
            }
            return count;
        }

        public void Dispose()
        {
            _channel.Dispose();
            _connection.Dispose();
        }

        private void InitClient()
        {
            _connection = _connectionChannelPool.GetConnection();

            _channel = _connection.CreateModel();

            //重试队列
            string eReply = $"{_exchageName}@reply";
            string qReply = $"{this.RabbitMQItem.QueueName}@reply";
            string rReply = $"{this.RabbitMQItem.RouteKey}@reply";
            _channel.ExchangeDeclare(eReply, _exchageType, true);
            var argsReply = new Dictionary<string, object>
            {
                {"x-message-ttl",1000*3},
                {"x-dead-letter-exchange", _exchageName},
                {"x-dead-letter-routing-key", this.RabbitMQItem.RouteKey}
            };
            _channel.QueueDeclare(qReply, true, false, false, argsReply);
            _channel.QueueBind(qReply, eReply, rReply);

            //失败队列
            string eFailed = $"{_exchageName}@failed";
            string qFailed = $"{this.RabbitMQItem.QueueName}@failed";
            string rFailed = $"{this.RabbitMQItem.RouteKey}@failed";
            _channel.ExchangeDeclare(eFailed, _exchageType, true);
            var argsFailed = new Dictionary<string, object>
            {
                {"x-message-ttl", _rabbitMQOptions.QueueMessageExpires}
            };
            _channel.QueueDeclare(qFailed, true, false, false, argsFailed);
            _channel.QueueBind(qFailed, eFailed, rFailed);

            //主队列
            _channel.ExchangeDeclare(_exchageName, _exchageType, true);
            var args = new Dictionary<string, object>
            {
                {"x-message-ttl", _rabbitMQOptions.QueueMessageExpires},
                {"x-max-priority",10 }
            };
            _channel.QueueDeclare(_queueName, true, false, false, args);
            _channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
        }

        #region events

        private void OnConsumerConsumerCancelled(object sender, ConsumerEventArgs e)
        {
            var args = new LogMessageEventArgs
            {
                LogType = MqLogType.ConsumerCancelled,
                Reason = e.ConsumerTags
            };
            OnLog?.Invoke(sender, args);
        }

        private void OnConsumerUnregistered(object sender, ConsumerEventArgs e)
        {
            var args = new LogMessageEventArgs
            {
                LogType = MqLogType.ConsumerUnregistered,
                Reason = e.ConsumerTags
            };
            OnLog?.Invoke(sender, args);
        }

        private void OnConsumerRegistered(object sender, ConsumerEventArgs e)
        {
            var args = new LogMessageEventArgs
            {
                LogType = MqLogType.ConsumerRegistered,
                Reason = e.ConsumerTags
            };
            OnLog?.Invoke(sender, args);
        }

        private void OnConsumerReceived(object sender, BasicDeliverEventArgs e)
        {
            _deliveryTag = e.DeliveryTag;
            _basicProperties = e.BasicProperties;
            var message = new MessageContext
            {
                Group = _queueName,
                Name = e.RoutingKey,
                Content = System.Text.Encoding.UTF8.GetString(e.Body.ToArray()) 
            };
            OnMessageReceived?.Invoke(sender, message);
        }

        private void OnConsumerShutdown(object sender, ShutdownEventArgs e)
        {
            var args = new LogMessageEventArgs
            {
                LogType = MqLogType.ConsumerShutdown,
                Reason = new string[] { e.ReplyText }
            };
            OnLog?.Invoke(sender, args);
        }

        #endregion
    }
}
